package me.shiqiang.im.linsteners;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.annotation.OnEvent;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.tst.common.utils.JSONUtils;
import me.shiqiang.im.enums.MessageCType;
import me.shiqiang.im.enums.MessageCmd;
import me.shiqiang.im.enums.MessageMType;
import me.shiqiang.im.msg.DistributeMsgBody;
import me.shiqiang.im.msg.MsgBody;
import me.shiqiang.im.msg.NewMsg;
import me.shiqiang.im.msg.SndMsg;
import me.shiqiang.im.redis.RedisKey;
import me.shiqiang.im.server.ConnectionContext;
import me.shiqiang.im.service.GroupClient;
import me.shiqiang.im.service.GroupService;
import me.shiqiang.im.service.NotifyService;
import me.shiqiang.im.util.RedisPipelineProcessor;
import me.shiqiang.im.utils.JsonUtils;
import me.shiqiang.im.utils.SnowFlake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author cuishiqiang
 * @date 2017/12/14 下午3:10
 */
@Component("SndMsgListener")
public class SndMsgListener extends Listener {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    StringRedisTemplate redisTemplate;

    @Autowired
    Producer producer;

    @Value("${server.id}")
    private String serverId;

    @Value("${com.tst.rocketmq.producer.topic}")
    private String topic;

    @Autowired
    ConcurrentLinkedQueue<RedisPipelineProcessor> queue;

    @Autowired
    GroupClient groupClient;

    @Autowired
    GroupService groupService;

    @Autowired
    SnowFlake snowFlake;

    @OnEvent(MessageCmd.SND_MSG)
    public void onSndMsg(SocketIOClient client, SndMsg sndMsg) throws JsonProcessingException {
//        ConnectionContext.getInstance().nextProcessor().sendEvent(MessageCmd.SND_MSG, sndMsg);
//        logger.info(JsonUtils.getInstance().toJson(sndMsg));


        long syncKey = snowFlake.nextId();

        String cType = String.valueOf(sndMsg.getcType());
        if (MessageCType.P2P.equals(cType)) {
            MsgBody msgBody = sndMsg.newMsgBody(syncKey, 0);
            String json = JsonUtils.getInstance().toJson(msgBody);
            queue.add(connection -> {
                connection.zAdd(RedisKey.clientMsgQueue(sndMsg.getFrom().toString()).getBytes(), syncKey, json.getBytes());
                connection.zAdd(RedisKey.clientMsgQueue(sndMsg.getTo().toString()).getBytes(), syncKey, json.getBytes());
                NewMsg fromNewMsg = new NewMsg(sndMsg.getFrom(), msgBody.getSyncKey(), msgBody);
                client.sendEvent(MessageCmd.NEW_MSG, fromNewMsg);
                NewMsg toNewMsg = new NewMsg(sndMsg.getTo(), msgBody.getSyncKey(), msgBody);

//                client.sendEvent(MessageCmd.NEW_MSG, toNewMsg);
                sendToClient(toNewMsg);
            });
        } else if (MessageCType.GROUP.equals(cType) || MessageCType.SERVICE_PLATFORM.equals(cType)) {
            sendGroupMsg(client, sndMsg, syncKey);

        }
    }

    private void sendGroupMsg(SocketIOClient client, SndMsg sndMsg, long syncKey) {
        //此处将Long[] at 原封不动传给分发端
        DistributeMsgBody distributeMsgBody = sndMsg.newMsgBody(syncKey);
        //添加对服务平台类型的支持,统一当群聊消息处理
        boolean muted = groupClient.isMuted(sndMsg.getTo(), sndMsg.getFrom());
        boolean allMutedAndIsNormal = groupClient.allMutedAndIsNormal(sndMsg.getTo(), sndMsg.getFrom());
        if (allMutedAndIsNormal) {
            return;
        }
        if (muted) {
            return;
        }
        Long groupId = distributeMsgBody.getTo();
        List<Long> userIds = groupService.members(groupId);
        Set<Long> atUserIds = new HashSet<>();
        boolean atAll = false;
        MsgBody toClientMsgBody = new MsgBody().from(distributeMsgBody);

        for (Long at : distributeMsgBody.getAt()) {
            if (at == -1) { // -1代表所有人
                atAll = true;
                break;
            }
            atUserIds.add(at);
        }
        boolean finalAtAll = atAll;

        if (finalAtAll) {
            toClientMsgBody.setAt(1);
        }

        queue.add(connection -> {
            for (Long userId : userIds) {
//                            if (!chatSwitchService.chatSwitch(Long.valueOf(userId))) {
//                                continue;
//                            }
                if (!atUserIds.isEmpty()) {
                    if (atUserIds.contains(userId)) {
                        toClientMsgBody.setAt(1);
                    } else {
                        toClientMsgBody.setAt(0);
                    }
                }

                NewMsg fromNewMsg = new NewMsg(userId, syncKey, toClientMsgBody);
                String userMsgMq = RedisKey.clientMsgQueue(String.valueOf(userId));
                byte[] toClientMsgBytes = new byte[0];
                try {
                    toClientMsgBytes = JSONUtils.toJSON(toClientMsgBody).getBytes();
                } catch (JsonProcessingException e) {
                    logger.error(String.format("消息格式转换异常，消息: %s", toClientMsgBody.toString()), e);
                }
                if (redisTemplate.opsForZSet().zCard(userMsgMq) == null || redisTemplate.opsForZSet().zCard(userMsgMq) == 0) {
                    connection.zAdd(userMsgMq.getBytes(), syncKey, toClientMsgBytes);
                    connection.expire(userMsgMq.getBytes(), 604800);//7天
                } else {
                    connection.zAdd(userMsgMq.getBytes(), syncKey, toClientMsgBytes);
                }
//                client.sendEvent(MessageCmd.NEW_MSG, fromNewMsg);
                sendToClient(fromNewMsg);
            }
        });
    }




    void sendToClient(NewMsg newMsg) {
        Long userId = newMsg.getUserId();
        String clientId = ConnectionContext.getInstance().get(userId.toString());
        try {
            MsgBody msgBody = newMsg.getMsgBody();
            if (clientId == null) {
                if (MessageMType.NOTICE.equalsIgnoreCase(String.valueOf(msgBody.getmType()))) {
                    return;
                }
                logger.info("clientId为null,{}已发送通知消息", JsonUtils.getInstance().toJson(newMsg));
//                sendMqMsg(String.valueOf(msgBody.getcType()), JsonUtils.getInstance().toJson(newMsg));
                sendNotification(userId.toString(), newMsg);
                return;
            }
            SocketIOClient userClient = server.getClient(UUID.fromString(clientId));
            if (userClient == null) {
                if (MessageMType.NOTICE.equalsIgnoreCase(String.valueOf(msgBody.getmType()))) {
                    return;
                }
                logger.info("userClient为null,已发送通知消息");
//                sendMqMsg(String.valueOf(msgBody.getcType()), JsonUtils.getInstance().toJson(newMsg));
                sendNotification(userId.toString(), newMsg);
                logger.info(JsonUtils.getInstance().toJson(newMsg));
                return;
            }

//            logger.info(String.format("clientId为%s,userClient:%s",clientId,userClient));
            userClient.sendEvent(MessageCmd.NEW_MSG, newMsg);
        } catch (IOException e) {
            logger.error("发生异常", e);
        }
    }

    @Autowired
    NotifyService notifyService;

    private void sendNotification(String userId,NewMsg newMsg) {
        notifyService.send(newMsg);
    }

    private void sendMqMsg(String cType, String msgStr) {
        Message msg = new Message(topic,cType,msgStr.getBytes());
        producer.sendOneway(msg);
    }
}
